package com.dwj.mastersel;

import org.I0Itec.zkclient.IZkDataListener;
import org.I0Itec.zkclient.ZkClient;
import org.I0Itec.zkclient.exception.ZkException;
import org.I0Itec.zkclient.exception.ZkInterruptedException;
import org.I0Itec.zkclient.exception.ZkNoNodeException;
import org.I0Itec.zkclient.exception.ZkNodeExistsException;
import org.apache.zookeeper.CreateMode;

import java.util.concurrent.Executors;
import java.util.concurrent.ScheduledExecutorService;
import java.util.concurrent.TimeUnit;

/**
 * Created by Administrator on 2017/11/14 0014.
 */
public class WorkServer {

    private volatile boolean running = false;

    private ZkClient zkClient;

    private static final String MASTER_PATH = "/master";

    private IZkDataListener dataListener;//监听master的删除事件

    private RunningData serverData;//记录当前服务器的基本信息

    private RunningData masterData; //记录master节点的基本信息

    private ScheduledExecutorService delayExector = Executors.newScheduledThreadPool(1);
    private int delatTime = 5;

    public WorkServer(RunningData rd){
        this.serverData = rd;
        dataListener = new IZkDataListener() {
            public void handleDataChange(String s, Object o) throws Exception {
                if(masterData!=null && masterData.getName().equals(serverData.getName())){
                    takeMaster();;
                }else {
                    delayExector.schedule(new Runnable() {
                        public void run() {
                            if(checkMaster()) {
                                takeMaster();
                            }
                        }
                    }, delatTime, TimeUnit.SECONDS);
                }
            }

            public void handleDataDeleted(String s) throws Exception {

            }
        };
    }
    public void start() throws Exception{
        if(running){
            throw new Exception("server has startup...");
        }
        running = true;
        zkClient.subscribeDataChanges(MASTER_PATH, dataListener);
        takeMaster();
    }

    public void stop() throws Exception{
        if(!running){
            throw new Exception("server has stoped...");
        }
        running = false;
        zkClient.unsubscribeDataChanges(MASTER_PATH, dataListener);
        releaseMaster();
    }

    //争抢master权利
    private void takeMaster(){
        if(!running){
            return;
        }
        try {
            zkClient.create(MASTER_PATH, serverData, CreateMode.EPHEMERAL);
            masterData = serverData;
        }catch (ZkNodeExistsException e){
            RunningData runningData = zkClient.readData(MASTER_PATH, true);
            if(null == runningData){
                takeMaster();
            }else {
                masterData = serverData;
            }
        }catch (Exception e){
            //ignore
        }
    }

    private void releaseMaster(){
        if(checkMaster()){
            zkClient.deleteRecursive(MASTER_PATH);
        }
    }

    private boolean checkMaster(){
        try {
            RunningData eventData = zkClient.readData(MASTER_PATH);
            masterData = eventData;
            if(masterData.getName().equals(serverData.getName())){
                return true;
            }
        }catch (ZkNoNodeException e){
            return false;
        }catch (ZkInterruptedException e){
            checkMaster();
        }catch (ZkException e){
            return false;
        }
        return false;
    }
}
